package cn.edu.xmu.datamining.tangzk.mralgos.kmeans;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Iterator;
import java.util.Random;

import org.apache.hadoop.io.WritableComparable;

/**
 * sample format
 * 
 * @version 2013-8-4
 * @author tangzk
 * @Reviewer
 * 
 */
public class SampleWritable implements WritableComparable<SampleWritable> {

	private float x1;
	private float x2;
	private float x3;

	public SampleWritable() {
	}

	public SampleWritable(float x1, float x2, float x3) {
		this.x1 = x1;
		this.x2 = x2;
		this.x3 = x3;
	}

	public static SampleWritable parseSample(String str) {
		SampleWritable sample = null;
		String[] flds = str.split(",");
		if (flds.length >= 3) {
			sample = new SampleWritable();
			sample.x1 = Float.parseFloat(flds[0]);
			sample.x2 = Float.parseFloat(flds[1]);
			sample.x3 = Float.parseFloat(flds[2]);
		}
		return sample;
	}

	public static SampleWritable newRandom() {
		Random rand = new Random();
		float x1 = rand.nextFloat();
		float x2 = rand.nextFloat();
		float x3 = rand.nextFloat();
		return new SampleWritable(x1, x2, x3);
	}

	/**
	 * get the distance from this to sample.
	 */
	public double getDistance(SampleWritable sample) {
		if (sample == null) {
			return Double.MAX_VALUE;
		} else if (this == sample) {
			return 0f;
		} else {
			SampleWritable impl = (SampleWritable) sample;
			return Math.sqrt((this.x1 - impl.x1) * (this.x1 - impl.x1)
					+ (this.x2 - impl.x2) * (this.x2 - impl.x2)
					+ (this.x3 - impl.x3) * (this.x3 - impl.x3));
		}
	}

	/**
	 * get the closest cluster center from this.
	 */
	public ClusterCentroidWritable getClosestCluster(
			Iterable<ClusterCentroidWritable> centroids) {
		ClusterCentroidWritable closestCentroid = null;
		if (centroids != null) {
			double mind = Double.MAX_VALUE;
			Iterator<ClusterCentroidWritable> iter = centroids.iterator();
			ClusterCentroidWritable centroid = null;
			double dist;
			while (iter.hasNext()) {
				centroid = iter.next();
				dist = this.getDistance(centroid.getSample());
				if (dist < mind) {
					closestCentroid = centroid;
					mind = dist;
				}
			}
		}
		return closestCentroid;
	}

	public static SampleWritable getMean(Iterable<SampleWritable> values) {
		return getMean(null, values);
	}

	/**
	 * get the mean sample from samples.
	 */
	public static SampleWritable getMean(SampleWritable initSample,
			Iterable<SampleWritable> values) {
		SampleWritable sum = null;
		int n = 0;
		if (initSample == null) {
			sum = new SampleWritable();
			n = 0;
		} else {
			sum = initSample;
			n = 1;
		}
		Iterator<SampleWritable> valIter = values.iterator();
		SampleWritable sample = null;
		while (valIter.hasNext()) {
			sample = (SampleWritable) valIter.next();
			sum.setX1(sum.getX1() + sample.getX1());
			sum.setX2(sum.getX2() + sample.getX2());
			sum.setX3(sum.getX3() + sample.getX3());
			n++;
		}
		sum.setX1(sum.getX1() / n);
		sum.setX2(sum.getX2() / n);
		sum.setX3(sum.getX3() / n);
		return sum;
	}

	@Override
	public void write(DataOutput out) throws IOException {
		out.writeFloat(this.x1);
		out.writeFloat(this.x2);
		out.writeFloat(this.x3);
	}

	@Override
	public void readFields(DataInput in) throws IOException {
		this.x1 = in.readFloat();
		this.x2 = in.readFloat();
		this.x3 = in.readFloat();
	}

	@Override
	public int compareTo(SampleWritable o) {
		if (this.x1 != o.x1) {
			return Float.compare(this.x1, o.x1);
		} else if (this.x2 != o.x2) {
			return Float.compare(this.x2, o.x2);
		} else {
			return Float.compare(this.x3, o.x3);
		}
	}

	@Override
	public boolean equals(Object obj) {
		if (!(obj instanceof SampleWritable)) {
			return false;
		}
		SampleWritable o = (SampleWritable) obj;
		return this.x1 == o.x1 && this.x2 == o.x2
				&& Float.compare(this.x3, o.x3) == 0;
	}

	@Override
	public int hashCode() {
		return (int) this.x1 * 31 + (int) this.x2 * 17 + (int) this.x3;
	}

	@Override
	protected SampleWritable clone() {
		return new SampleWritable(x1, x2, x3);
	}

	@Override
	public String toString() {
		return this.x1 + "," + this.x2 + "," + this.x3;
	}

	public float getX1() {
		return x1;
	}

	public void setX1(float x1) {
		this.x1 = x1;
	}

	public float getX2() {
		return x2;
	}

	public void setX2(float x2) {
		this.x2 = x2;
	}

	public float getX3() {
		return x3;
	}

	public void setX3(float x3) {
		this.x3 = x3;
	}

}
